package org.example.watermark;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;

public class MyPuntuatedWatermarkGenerator<T> implements WatermarkGenerator<T> {
    private static final long serialVersionUID = 1L;
    //乱序等待时间
    private Long delayTs;
    //当前最大的Watermark时间
    private Long maxTs;

    MyPuntuatedWatermarkGenerator(Long delayTs) {
        this.delayTs = delayTs;
        this.maxTs = Long.MIN_VALUE + delayTs + 1;
    }

    /**
     * @param event
     * @param eventTimestamp 事件时间
     * @param output
     */
    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTs = Math.max(this.maxTs, eventTimestamp);
        output.emitWatermark(new Watermark(maxTs - delayTs - 1));
        System.out.println("调用onEvent，取得最大的watermark时间戳：" + maxTs + "，生成watermark时间戳：" + (maxTs - delayTs - 1));
    }

    /**
     * 周期性的生成Watermark
     *
     * @param output
     */
    @Override
    public void onPeriodicEmit(WatermarkOutput output) {

    }
}
